package com.winning.platwebservice.controller;

import com.winning.platwebservice.service.Impl.KafkaServiceImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Objects;

/**
 * @Author: xkh@winning.com.cn
 * @Desc: Kafka用于消息传输
 * @Date: 2019-03-08 10:31
 **/
@Component
@ConfigurationProperties(prefix = "kafka.producer.send", ignoreInvalidFields = true, ignoreUnknownFields = true)
public class KafkaCollect {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private KafkaServiceImpl kafkaService;

    private List<String> topics = new ArrayList<String>();

    private Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 发送消息到Kafka主题
     *
     * @param data 要发送的数据
     */
    public void producer(Object data) {
        String message = serializeObject(data);
        if (Objects.isNull(message)) {
            return;
        }
        topics.stream().forEach(topic -> {
            kafkaTemplate.send(topic, message);
        });
    }

    /**
     * 消费数据
     * topics：配置数据来源的主题
     *
     * @param source 消费的源数据
     */
    @KafkaListener(topics = "test", groupId = "0")
//    @KafkaListener(topics = "#{new String('${kafka.consumer.listener.topics}').split(',')}")
    public void consumer(String source) {
        Object data = deserializeObject(source);
        if (Objects.isNull(data)) {
            return;
        }
        kafkaService.processUpdate(data);
    }

    /**
     * 序列化对象
     *
     * @param object 序列化的对象
     * @return Base64编码后的字符串
     */
    protected String serializeObject(Object object) {
        try {
            ByteArrayOutputStream byteArrayOutputStream =
                    new ByteArrayOutputStream();
            ObjectOutputStream objectOutputStream =
                    new ObjectOutputStream(byteArrayOutputStream);
            objectOutputStream.writeObject(object);
            byte[] bytes = byteArrayOutputStream.toByteArray();

            objectOutputStream.close();
            byteArrayOutputStream.close();

            return Base64.getEncoder().encodeToString(bytes);
        } catch (IOException e) {
            logger.error("Kafka数据采集-序列化对象出现错误", e);
        }

        return null;
    }

    /**
     * 反序列化对象，重新构建对象
     *
     * @param source 序列化并Base64编码后的字符串
     * @return 重新构建的对象
     */
    protected Object deserializeObject(String source) {
        try {
            byte[] bytes = Base64.getDecoder().decode(source);
            ByteArrayInputStream byteArrayInputStream =
                    new ByteArrayInputStream(bytes);
            ObjectInputStream objectInputStream =
                    new ObjectInputStream(byteArrayInputStream);

            Object object = objectInputStream.readObject();

            objectInputStream.close();
            byteArrayInputStream.close();

            return object;
        } catch (Exception e) {
            logger.error("Kafka数据采集-反序列化对象出现错误", e);
        }

        return null;
    }


    public List<String> getTopics() {
        return topics;
    }
}
